package priv.pfz.raft.peer;

import com.google.common.collect.Lists;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import priv.pfz.raft.network.RpcManager;
import priv.pfz.raft.network.dto.*;
import priv.pfz.raft.peer.enums.OperateType;
import priv.pfz.raft.peer.enums.PeerState;
import priv.pfz.raft.peer.model.*;

import java.util.Map;

import static priv.pfz.raft.peer.enums.Actions.*;

/**
 * @author pengfangzhou
 * @date 2022/2/3 18:11
 */
@Getter
@Setter
@Slf4j
public class Peer {
    private PeerConf peerConf;
    private PeerData peerData;

    private static final int LOG_BATCH_NUM = 5;

    /**
     * 获取状态
     */
    public String status() {
        return peerData.toString();
    }

    /**
     * 暂停
     */
    public synchronized void pause() {
        peerData.getElectionQueue().pause();
        for (OtherPeerInfo otherPeerInfo : peerData.getOtherPeerInfoMap().values()) {
            otherPeerInfo.getRpcQueue().pause();
        }
    }

    /**
     * 恢复
     */
    public synchronized void resume() {
        peerData.getElectionQueue().resume();
        for (OtherPeerInfo otherPeerInfo : peerData.getOtherPeerInfoMap().values()) {
            otherPeerInfo.getRpcQueue().resume();
        }
    }

    /**
     * 启动
     */
    public synchronized void start() {
        if (peerData.getState() != PeerState.STOPPED) {
            log.info("{} is running now", peerConf.getPeerName());
            return;
        }
        log.info("{}启动[{}]", peerConf.getPeerName(), peerData.getCurrentTerm());
        log.debug("初始状态:{}", peerData);
        becomeFollower(null, null);
        log.debug("结束状态:{}", peerData);
    }

    /**
     * 成为follower
     */
    private void becomeFollower(String leader, Integer term) {
        if (term != null && term > peerData.getCurrentTerm()) {
            peerData.clearForTermUpdate(term);
        } else {
            peerData.clearForStateUpdate();
        }

        //状态切换为follower
        peerData.setState(PeerState.FOLLOWER);
        peerData.setCurrentLeader(leader);

        //设置选举计时器，超时未收到心跳或未投票则进入选举
        int maxTimeout = peerConf.getMaxElectionTimeout();
        int timeout = RandomUtils.nextInt(maxTimeout / 2, maxTimeout);
        peerData.getElectionQueue().add(SWITCH_TO_CANDIDATE, this::switchToCandidate, timeout);
    }

    /**
     * 停止
     */
    public synchronized void stop() {
        if (peerData.getState() == PeerState.STOPPED) {
            log.info("{} is already stopped", peerConf.getPeerName());
            return;
        }
        log.info("{}停止", peerConf.getPeerName());
        log.debug("初始状态:{}", peerData);

        peerData.clearForStateUpdate();

        //状态切换为stopped
        peerData.setState(PeerState.STOPPED);

        log.debug("结束状态:{}", peerData);
    }

    /**
     * 成为candidate
     */
    private synchronized void switchToCandidate() {
        int term = peerData.getCurrentTerm() + 1;
        log.info("{}切换为candidate[{}]", peerConf.getPeerName(), term);
        log.debug("初始状态:{}", peerData);

        peerData.clearForTermUpdate(term);

        //状态切换为candidate，并向自己投票
        peerData.setState(PeerState.CANDIDATE);
        peerData.setVotedFor(peerConf.getPeerName());
        peerData.incrGrantedCount();

        //设置选举计时器，超时未选举成功或选举失败，则发起新一轮选举
        int maxTimeout = peerConf.getMaxElectionTimeout();
        int timeout = RandomUtils.nextInt(maxTimeout / 2, maxTimeout);
        peerData.getElectionQueue().add(SWITCH_TO_CANDIDATE, this::switchToCandidate, timeout);

        //向其它节点首次拉票
        for (Map.Entry<String, OtherPeerInfo> entry : peerData.getOtherPeerInfoMap().entrySet()) {
            String peer = entry.getKey();
            sendRequestVote(peer);
        }

        log.debug("结束状态:{}", peerData);
    }

    private synchronized void sendRequestVote(String peer) {
        //设置下次拉票
        long interval = peerConf.getRpcInterval();
        peerData.getOtherPeerInfo(peer).getRpcQueue().add(SEND_REQUEST_VOTE, () -> sendRequestVote(peer), interval);

        RequestVoteReq req = new RequestVoteReq();
        req.setSender(peerConf.getPeerName());
        req.setReceiver(peer);
        req.setTerm(peerData.getCurrentTerm());
        req.setCandidateId(peerConf.getPeerName());
        req.setLastLogKey(peerData.getLogs().getLastLogKey());

        log.info("{}向{}发起拉票:{}", req.getSender(), req.getReceiver(), req);
        RpcManager.requestVote(req, (requestVoteResp) -> handleRequestVoteResp(req, requestVoteResp));
    }

    /**
     * 处理拉票请求
     */
    public synchronized RequestVoteResp handleRequestVote(RequestVoteReq req) {
        if (peerData.getState() == PeerState.STOPPED) {
            return null;
        }

        log.info("{}处理{}的拉票请求:{}", req.getReceiver(), req.getSender(), req);
        log.debug("初始状态:{}", peerData);

        RequestVoteResp resp = new RequestVoteResp();
        resp.setTerm(peerData.getCurrentTerm());
        resp.setVoteGranted(canVote(req));

        log.debug("结束状态:{}", peerData);
        log.info("响应:{}", resp);
        return resp;
    }

    private boolean canVote(RequestVoteReq req) {
        //拉票term小于当前term，返回false
        if (req.getTerm() < peerData.getCurrentTerm()) {
            return false;
        }
        //拉票者日志比本地少，返回false
        LogKey localLog = peerData.getLogs().getLastLogKey();
        LogKey candidateLog = req.getLastLogKey();
        if (candidateLog.compareTo(localLog) < 0) {
            return false;
        }
        //本期已投票，根据投给的人返回true/false
        if (req.getTerm().equals(peerData.getCurrentTerm()) && peerData.getVotedFor() != null) {
            return peerData.getVotedFor().equals(req.getCandidateId());
        }
        //若拉票者term比本地大，无条件变成follower
        if (req.getTerm() > peerData.getCurrentTerm()) {
            becomeFollower(null, req.getTerm());
        }
        //投票给这个拉票者
        peerData.setVotedFor(req.getCandidateId());

        return true;
    }

    /**
     * 处理拉票请求的响应
     */
    private synchronized void handleRequestVoteResp(RequestVoteReq req, RequestVoteResp resp) {
        if (peerData.getState() == PeerState.STOPPED) {
            return;
        }

        //当前不再是竞选者，或者不是当前任期，无需处理
        if (peerData.getState() != PeerState.CANDIDATE || peerData.getCurrentTerm() != req.getTerm()) {
            return;
        }

        log.info("{}处理{}的拉票响应:{}", req.getSender(), req.getReceiver(), resp);
        log.debug("初始状态:{}", peerData);

        //记住这个选民已经投了票，不再发拉票请求
        OtherPeerInfo otherPeerInfo = peerData.getOtherPeerInfo(req.getReceiver());
        otherPeerInfo.setVoteGranted(resp.getVoteGranted());
        otherPeerInfo.getRpcQueue().clear();
        if (resp.getVoteGranted()) {
            peerData.incrGrantedCount();
        } else {
            peerData.incrDeniedCount();
        }

        log.info("当前同意票数{},拒绝票数{}", peerData.getGrantedCount(), peerData.getDeniedCount());

        //投票已经明确失败，关闭所有拉票，等下一轮竞选
        if (peerData.getDeniedCount() >= peerConf.getMajorityNum()) {
            for (OtherPeerInfo peerInfo : peerData.getOtherPeerInfoMap().values()) {
                peerInfo.getRpcQueue().clear();
            }
        }

        //总共拿到超过半数选票，变成leader
        if (peerData.getGrantedCount() >= peerConf.getMajorityNum()) {
            becomeLeader();
        }

        log.debug("结束状态:{}", peerData);
    }

    /**
     * 成为leader
     */
    private void becomeLeader() {
        log.info("{}切换为leader[{}]", peerConf.getPeerName(), peerData.getCurrentTerm());

        peerData.clearForStateUpdate();

        //状态切换为leader
        peerData.setState(PeerState.LEADER);

        //向其它节点首次发送复制日志
        for (Map.Entry<String, OtherPeerInfo> entry : peerData.getOtherPeerInfoMap().entrySet()) {
            String peer = entry.getKey();
            sendAppendEntries(peer);
        }
    }

    /**
     * 发起复制日志请求
     */
    private synchronized void sendAppendEntries(String peer) {
        OtherPeerInfo otherPeerInfo = peerData.getOtherPeerInfo(peer);

        //设置下次心跳
        long interval = peerConf.getRpcInterval();
        otherPeerInfo.getRpcQueue().add(SEND_APPEND_ENTRIES, () -> sendAppendEntries(peer), interval);
        //将该节点设置为正在连接状态
        otherPeerInfo.setConnecting(true);

        AppendEntriesReq req = buildAppendEntriesReq(peer);
        log.info("{}向{}发起日志复制:{}", req.getSender(), req.getReceiver(), req);
        RpcManager.appendEntries(req, (appendEntriesResp) -> handleAppendEntriesResp(req, appendEntriesResp));
    }

    private AppendEntriesReq buildAppendEntriesReq(String peerName) {
        AppendEntriesReq req = new AppendEntriesReq();
        req.setSender(peerConf.getPeerName());
        req.setReceiver(peerName);
        req.setTerm(peerData.getCurrentTerm());
        req.setLeaderId(peerConf.getPeerName());
        req.setCommitLogKey(peerData.getLogs().getCommitLogKey());

        if (peerData.hasDiffLog(peerName)) {
            //单条发送
            LogKey maxLogKey = peerData.getOtherPeerInfo(peerName).getMaxLogKey();
            LogEntry logEntry = peerData.getLogs().getNextLog(maxLogKey);
            req.setPrevLogKey(maxLogKey);
            req.setLogEntries(Lists.newArrayList(logEntry));
        }
        return req;
    }

    /**
     * 处理日志复制请求
     */
    public synchronized AppendEntriesResp handleAppendEntries(AppendEntriesReq req) {
        if (peerData.getState() == PeerState.STOPPED) {
            return null;
        }

        log.info("{}处理{}的日志复制请求:{}", req.getReceiver(), req.getSender(), req);
        log.debug("初始状态:{}", peerData);

        AppendEntriesResp resp = new AppendEntriesResp();
        resp.setTerm(peerData.getCurrentTerm());
        resp.setSuccess(appendEntries(req));
        resp.setMaxLogKey(peerData.getLogs().getLastLogKey());

        log.debug("结束状态:{}", peerData);
        log.info("响应:{}", resp);
        return resp;
    }

    private boolean appendEntries(AppendEntriesReq req) {
        //请求term小于当前term，丢弃
        if (req.getTerm() < peerData.getCurrentTerm()) {
            return false;
        }

        //更新本地状态
        becomeFollower(req.getLeaderId(), req.getTerm());

        //请求term大于当前term，删除本地未提交的日志
        if (req.getTerm() > peerData.getCurrentTerm()) {
            LogKey commitLog = peerData.getLogs().getCommitLogKey();
            peerData.getLogs().getLogs().removeIf(logEntry -> logEntry.getLogKey().compareTo(commitLog) > 0);
        }

        //写日志
        boolean success = followerWriteLogs(req);

        //提交日志
        followerCommitLog(req);

        return success;
    }

    private boolean followerWriteLogs(AppendEntriesReq req) {
        if (req.getLogEntries().isEmpty()) {
            return true;
        }

        //复制的日志和本地日志不连续，返回写入失败
        LogKey localLog = peerData.getLogs().getLastLogKey();
        LogKey reqPrevLog = req.getPrevLogKey();
        if (localLog.compareTo(reqPrevLog) != 0) {
            return false;
        }

        //本地写日志
        LogEntry logEntry = req.getLogEntries().get(0);
        peerData.getLogs().addLog(logEntry);
        return true;
    }

    private void followerCommitLog(AppendEntriesReq req) {
        //leader提交进度 > 本地日志进度
        LogKey localLog = peerData.getLogs().getLastLogKey();
        LogKey leaderCommitLog = req.getCommitLogKey();
        if (leaderCommitLog.compareTo(localLog) > 0) {
            return;
        }
        //leader提交进度 <= 本地提交进度
        LogKey localCommitLog = peerData.getLogs().getCommitLogKey();
        if (leaderCommitLog.compareTo(localCommitLog) <= 0) {
            return;
        }
        peerData.getLogs().setCommitLogKey(req.getCommitLogKey());
        LogEntry commitLog = peerData.getLogs().getLog(req.getCommitLogKey());
        if (commitLog.getOperateType() == OperateType.WRITE) {
            //写操作需要应用到状态机
            peerData.setCurrentValue(commitLog.getValue());
        }
    }

    /**
     * 处理复制日志的响应
     */
    private synchronized void handleAppendEntriesResp(AppendEntriesReq req, AppendEntriesResp resp) {
        log.info("{}处理{}的日志复制响应:{}", req.getSender(), req.getReceiver(), resp);
        log.debug("初始状态:{}", peerData);

        String peerName = req.getReceiver();
        OtherPeerInfo otherPeerInfo = peerData.getOtherPeerInfo(peerName);

        //将该节点还原为非连接状态
        otherPeerInfo.setConnecting(false);
        otherPeerInfo.setMaxLogKey(resp.getMaxLogKey());

        //检查是否有可提交的日志
        leaderCommitLog();

        otherPeerInfo.getRpcQueue().clear();
        if (peerData.hasDiffLog(peerName)) {
            //仍有差异则立即发送一个日志复制请求
            sendAppendEntries(peerName);
        } else {
            //暂无差异则设置日志复制的计时器
            long interval = peerConf.getRpcInterval();
            otherPeerInfo.getRpcQueue().add(SEND_APPEND_ENTRIES, () -> sendAppendEntries(peerName), interval);
        }

        log.debug("结束状态:{}", peerData);
    }

    private void leaderCommitLog() {
        LogEntry cur = peerData.getLogs().getNextLog(peerData.getLogs().getCommitLogKey());
        while (true) {
            if (cur == null) {
                break;
            }
            //计算已复制的节点数，包含自身
            LogKey curLogKey = cur.getLogKey();
            long writeCount = peerData.getOtherPeerInfoMap().values().stream()
                    .filter(otherPeerInfo -> otherPeerInfo.getMaxLogKey().compareTo(curLogKey) >= 0)
                    .count();
            writeCount++;
            //未超半数，退出
            if (writeCount < peerConf.getMajorityNum()) {
                break;
            }
            //本条日志超半数复制，进行提交，然后看下一条
            peerData.getLogs().setCommitLogKey(curLogKey);
            if (cur.getOperateType() == OperateType.WRITE) {
                //写操作需要应用到状态机
                peerData.setCurrentValue(cur.getValue());
            }
            cur = peerData.getLogs().getNextLog(cur.getLogKey());
        }
    }

    /**
     * 处理客户端本地读请求
     */
    public ReadResp handleLocalRead(ReadReq req) {
        ReadResp resp = new ReadResp();
        resp.setValue(peerData.getCurrentValue());
        return resp;
    }

    /**
     * 处理客户端读请求
     */
    public synchronized ReadResp handleRead(ReadReq req) {
        ReadResp resp = new ReadResp();
        LogEntry logEntry = new LogEntry();
        logEntry.setOperateType(OperateType.READ);
        resp.setMsg(doOperate(logEntry));
        return resp;
    }

    /**
     * 处理客户端写请求
     */
    public synchronized WriteResp handleWrite(WriteReq req) {
        WriteResp resp = new WriteResp();
        LogEntry logEntry = new LogEntry();
        logEntry.setOperateType(OperateType.WRITE);
        logEntry.setValue(req.getValue());
        resp.setMsg(doOperate(logEntry));
        return resp;
    }

    private String doOperate(LogEntry logEntry) {
        //当前不是主节点，拒绝操作
        if (peerData.getState() != PeerState.LEADER) {
            return "not leader";
        }

        //本地写入日志
        peerData.incrCurrentIndex();
        logEntry.setTerm(peerData.getCurrentTerm());
        logEntry.setIndex(peerData.getCurrentIndex());
        peerData.getLogs().addLog(logEntry);

        //重置向其它非正在连接的节点复制日志/心跳的计时器
        for (Map.Entry<String, OtherPeerInfo> entry : peerData.getOtherPeerInfoMap().entrySet()) {
            String peer = entry.getKey();
            OtherPeerInfo otherPeerInfo = entry.getValue();
            if (!otherPeerInfo.isConnecting()) {
                otherPeerInfo.getRpcQueue().clear();
                sendAppendEntries(peer);
            }
        }

        return "processing";
    }


}
